Description

This PR fixes partial-failure and recovery edge cases across the write path, Pipe, subscription progress, and ConfigNode snapshots. These bugs could replay data that had already been written, report unwritten rows as successful, leak generated TsFiles or reserved memory, or retain stale state after recovery.

Keep write retries and failure reporting consistent

• Track processed tablet rows across a type-inconsistency flush retry so only the remaining sequence/unsequence and time-partition fragments are retried.
• Preserve successful rows when retrying row batches, report processor-creation and insert failures at their original row indexes, and mark every uninserted tablet range with the original failure status.
• Roll back memory reserved by earlier aligned/non-aligned row groups or tablet ranges when a later reservation fails. This keeps memory accounting, last-cache updates, and consensus retry decisions consistent with the rows actually inserted.
• Keep recovered write metadata consistent by taking the maximum end time across TsFile resources and clearing per-partition memory accounting with the last-flush-time map.

Make Pipe conversion and transfer failure-safe

• Buffer table-model tablets into the V2 fallback builder and remove partial primary-path outputs before either V2 builder falls back. The tree-model builder also skips individual null schemas and accepts null DATE values without dropping valid columns or forcing fallback.
• Delete temporary TsFiles created before writer initialization, multi-database sealing, partial batch conversion, or sync/air-gap/async transfer failures.
• Complete the async error path when scheduling fails or a client has already returned to the pool, and release readers, memory blocks, event references, and generated batch files during handler or subscription cleanup. Source TsFiles remain untouched.

Preserve subscription commit progress safely

• Use versioned, encoded composite keys for consumer group, topic, and region identifiers so delimiters and numeric DataNode ID prefixes cannot collide. During rolling upgrades, write and merge unambiguous legacy keys while preferring the versioned form.
• Remove persisted progress when a consumer group unsubscribes from a topic, and advance the persistence throttle when broadcast progress is applied.
• Read snapshot fields fully: a completely absent commit-progress section remains compatible with older snapshots, while truncated headers or entries fail loading instead of silently losing progress.

Replace stale state during snapshot loading

• Replace all derived template indexes and quota usage state, and clear empty template-preset or external-service snapshots instead of retaining values created after the snapshot.
• Close listening queues before deserialization so old events and references are released, reset empty linked-queue indexes and links, and rebuild ConfigNode Pipe listener references from the loaded Pipe metadata.

Verification

• mvn spotless:apply -pl iotdb-core/node-commons,iotdb-core/datanode,iotdb-core/confignode
• Targeted unit tests: CommitProgressKeeperTest, QuotaInfoTest, TemplateTableTest, TemplatePreSetTableTest, PipeTsFileBuilderV2Test, DataRegionTest, and TsFileProcessorTest
• Additional regression coverage was added for ConfigNode commit-progress merging, Pipe listener and queue snapshot restoration, Pipe/subscription file cleanup, subscription commit state, external-service snapshots, and last-flush-time accounting.
• mvn -pl iotdb-core/node-commons,iotdb-core/datanode,iotdb-core/confignode -DskipTests -P with-zh-locale test-compile

This PR has:

• been self-reviewed.
• added comments explaining the "why" and the intent where it would not be obvious.
• added unit tests or modified existing tests to cover the new code paths.

Key changed/added classes (or packages if there are too many classes) in this PR

• DataRegion, TsFileProcessor, and HashLastFlushTimeMap
• Pipe TsFile builders, batches, sinks, and async transfer handlers
• CommitProgressKeeper, ConsensusSubscriptionCommitManager, and SubscriptionInfo
• TemplateTable, TemplatePreSetTable, QuotaInfo, and ExternalServiceInfo
• PipeConfigRegionListener, AbstractSerializableListeningQueue, and ConcurrentIterableLinkedQueue